/*
 * TcpClient.cpp
 *
 *  Created on: 2012-11-29
 *      Author: yangentao@gmail.com
 */

#include "TcpClient.h"
#include "../Log.h"
#include "../ThreadHandler.h"

namespace yet {

DWORD EventSelect::readThreadTask(void* argp) {
	DWORD result = 0;
	try {
		while (isSelected()) {
			DWORD dw = WSAWaitForMultipleEvents(1, &selectEvent, FALSE, WSA_INFINITE, FALSE);
			if (!isSelected()) {
				return result;
			}
			switch (dw) {
			case WSA_WAIT_FAILED : {
				SocketError e;
				Log << "WSAWaitForMultipleEvents Failed: " << e.getCode() << endl;
				throw e;
			}
				break;
			case WSA_WAIT_TIMEOUT: //never reach here , infinite
				break;
			case WSA_WAIT_IO_COMPLETION : //never reach here, fAlertable =false
				break;
			default: {
				WSANETWORKEVENTS e = { 0 };
				int n = WSAEnumNetworkEvents(sock->sock, selectEvent, &e);
				if (n != 0) { //failed
					throw SocketError();
				}
				long evt = e.lNetworkEvents;
				int* err = e.iErrorCode;

				if (evt & FD_READ) {
					int code = err[FD_READ_BIT];
					if (code) {
						throw SocketError(code);
					}
					String buf(READ_BUF_SIZE);
					int n = sock->recv(buf.buffer(), READ_BUF_SIZE);
					if (n > 0) {
						buf.setLength(n);
						OnSocketEvent(sock, Message(FD_READ, buf));
					} else if (n == 0) {
						sock->setInvalid();
						cancel();
						return result;
					} else {
						throw SocketError();
					}
				}

				if (evt & FD_OOB) {
					int code = err[FD_OOB_BIT];
					if (code) {
						throw SocketError(code);
					}
					String buf(READ_BUF_SIZE);
					int n = sock->recv(buf.buffer(), READ_BUF_SIZE, MSG_OOB);
					if (n > 0) {
						buf.setLength(n);
						OnSocketEvent(sock, Message(FD_OOB, buf));
					} else if (n == 0) {
						sock->setInvalid();
						cancel();
						return result;
					} else {
						throw SocketError();
					}
				}

				if (evt & FD_CONNECT) {
					int code = err[FD_CONNECT_BIT];
					if (code) {
						throw SocketError(code);
					}
					OnSocketEvent(sock, Message(FD_CONNECT));
				}
				if (evt & FD_WRITE) {
					int code = err[FD_WRITE_BIT];
					if (code) {
						throw SocketError(code);
					}
					OnSocketEvent(sock, Message(FD_WRITE));
				}
				if (evt & FD_CLOSE) {
					int code = err[FD_CLOSE_BIT];
					if (code) {
						throw SocketError(code);
					}
					sock->setInvalid();
					OnSocketEvent(sock, Message(FD_CLOSE));
					cancel();
					return result;
				}
				break;
			}

			}
		}
	} catch (SocketError& e) {
		e.print();
		cancel();
		result = e.getCode();
	}
	return result;
}

EventSelect::EventSelect() {
	selectEvent = 0;
	sock = 0;
	selected = false;
}
//TODO wait thread end
EventSelect::~EventSelect() {
	cancel();
}
HANDLE EventSelect::getHandle() {
	return selectEvent;
}
bool EventSelect::select(BaseSocket* s) {
	AutoLock al(this);
	if (!OnSocketEvent) {
		Log << err << "OnSocketEvent must set before invoking EventSelect::select.  " << endl;
		return false;
	}

	recvThread.OnRun = boost::bind(&EventSelect::readThreadTask, this, _1);
	if (recvThread.isAlive()) {
		recvThread.waitFinish();
	}
	if (recvThread.isAlive()) {
		Log << err << "recv thread already running." << endl;
		return false;
	}
	if (!recvThread.start()) {
		return false;
	}
	sock = s;
	long netEvents = FD_READ | FD_WRITE | FD_OOB | FD_ACCEPT | FD_CONNECT | FD_CLOSE;
	selectEvent = WSACreateEvent();
	int n = WSAEventSelect(sock->sock, selectEvent, netEvents);
	if (n != 0) {
		cancel();
		return false;
	}
	selected = true;
	return true;
}
void EventSelect::cancel() {
	selected = false;
	AutoLock al(this);
	if (selectEvent) {
		WSAEventSelect(sock->sock, 0, 0);
		WSASetEvent(selectEvent);
		WSACloseEvent(selectEvent);
		selectEvent = 0;
	}
	if (sock) {
		sock->close();
		sock = 0;
	}
	if (recvThread.getThreadID() != GetCurrentThreadId()) {
		if (recvThread.isAlive()) {
			recvThread.waitFinish();
		}
	}
}
bool EventSelect::isSelected() {
	return selected;
}

static void messageHandler(TcpClient* tcpClient, Message& msg) {
	if (tcpClient->OnEvent) {
		tcpClient->OnEvent(tcpClient, msg.msg, msg.argS);
	}
}

TcpClient::TcpClient() {
	init(0);
}
TcpClient::TcpClient(SOCKET s) {
	init(s);
}
TcpClient::~TcpClient() {
	if (status != stClosed && status != stInit) {
		close();
	}
//	msgHandler.quit();
	msgHandler.postQuitMessage();
	msgHandler.waitFinish();

}
void TcpClient::init(SOCKET s) {
	AutoLock al(this);
	status = stInit;
	sock.sock = s;
	msgHandler.addHandler(boost::bind(&messageHandler, this, _1));
	if (!msgHandler.start()) {
		Log << err << "TcpClient::init, Message Handler Start Failed!" << endl;
	}
	selector.OnSocketEvent = boost::bind(&TcpClient::onSocketEvent, this, _1, _2);
}

bool TcpClient::isWritable() {
	AutoLock al(this);
	return status == stConnected;
}

void TcpClient::onSocketEvent(BaseSocket* s, Message msg) {
	switch (msg.msg) {
	case FD_READ:
		break;
	case FD_OOB:
		break;
	case FD_WRITE:
		status = stConnected;
		break;
	case FD_CONNECT:
		status = stConnected;
		break;
	case FD_CLOSE:
		status = stClosed;
		break;
	}
	msgHandler.post(msg);
}

bool TcpClient::open(string remoteIP, short remotePort) {
	assert(status == stInit || status == stClosed);
	AutoLock al(this);
	status = stConnecting;
	if (!sock.createTcpSocket()) {
		status = stInit;
		return false;
	}
	if (!sock.bind()) {
		sock.close();
		status = stInit;
		return false;
	}

	if (!selector.select(&sock)) {
		status = stClosed;
		return false;
	}

	int n = sock.connect(remoteIP, remotePort);
	if (n == 0) {
		status = stConnected;
		return true;
	} else {
		SocketError e;
		if (e.isWouldBlock()) {
			status = stConnecting;
			return true;
		} else {
			status = stClosing;
			selector.cancel();
			status = stClosed;
			return false;
		}
	}

}
int TcpClient::send(String buf) {
	return send(buf.buffer(), buf.size());
}
int TcpClient::send(string buf) {
	return send(buf.c_str(), buf.size());
}

int TcpClient::send(const char* buf, int size) {
	AutoLock al(this);
	assert(isWritable());
	int n = sock.send(buf, size);
	if (n < 0) {
		SocketError e;
		if (e.isWouldBlock()) {
			status = stWaitWritable;
		} else {
			close();
		}
	}
	return n;
}
void TcpClient::close() {
	AutoLock al(this);

	status = stClosing;
	selector.cancel();
	status = stClosed;
}
TcpClient::Status TcpClient::getStatus() {
	return status;
}

} /* namespace yet */
